package com.edu.chat.web.shared;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.edu.buservice.util.model.TenantUtil;
import com.edu.chat.facade.facade.RecallFacade;
import com.edu.chat.facade.model.dto.RecallResDTO;
import com.edu.chat.facade.model.enums.SourceEnum;
import com.edu.chat.facade.model.request.RecallReq;
import com.edu.chat.web.common.RedisUtil;
import com.edu.chat.web.config.CommonConfig;
import com.edu.chat.web.covert.ArticleCovert;
import com.edu.chat.web.covert.DocumentCovert;
import com.edu.chat.web.dto.DocumentDTO;
import com.edu.chat.web.dto.LibraryInfoDTO;
import com.edu.chat.web.dto.SearchArticleDTO;
import com.edu.chat.web.dto.SearchDocDTO;
import com.edu.chat.web.integration.MilvusClient;
import com.edu.chat.web.model.CsArticle;
import com.edu.chat.web.model.CsDocuments;
import com.edu.chat.web.service.CsArticleService;
import com.edu.chat.web.service.CsDocumentsService;
import com.edu.chat.web.service.CsLibraryService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

@DubboService(interfaceClass = RecallFacade.class)
@Service
@Slf4j
public class RecallFacadeImpl implements RecallFacade {

	@Autowired
	private MilvusClient milvusClient;

	@Autowired
	private CsDocumentsService documentsService;

	@Autowired
	private CsArticleService articleService;

	@Autowired
	private CsLibraryService libraryService;

	@Autowired
	private RedisUtil redisUtil;

	private static final int CACHE_EXPIRE_TIME = 60 * 60;

	// 自定义线程池
	private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(
			5,
			20,
			60L,
			TimeUnit.SECONDS,
			new LinkedBlockingQueue<>(100),
			new ThreadPoolExecutor.CallerRunsPolicy());

	@Override
	public List<RecallResDTO> recall(RecallReq recallReq) {
		log.info("RecallFacadeImpl recall recallReq: {}", recallReq);

		if (recallReq == null || StringUtils.isBlank(recallReq.getContent())) {
			log.error("Invalid recall request: request object or content is empty");
			return Collections.emptyList();
		}

		try {
			List<String> publicLibraryIds = recallReq.getPublicLibraryIds();
			List<String> privateLibraryIds = recallReq.getPrivateLibraryIds();

			// 异步处理公共库和私有库
			CompletableFuture<List<RecallResDTO>> publicFuture = processLibrariesAsync(recallReq.getContent(), privateLibraryIds, recallReq.getTenantId());
			CompletableFuture<List<RecallResDTO>> privateFuture = processLibrariesAsync(recallReq.getContent(), publicLibraryIds, CommonConfig.PUBLIC_TENANT_ID);

			// 合并结果
			List<RecallResDTO> result = Stream.concat(publicFuture.join().stream(), privateFuture.join().stream())
					.sorted(Comparator.comparingDouble(RecallResDTO::getConfidence).reversed())
					.collect(Collectors.toList());

			// 处理截取数量
			int recallNum = recallReq.getRecallNum() == null ? 10 : recallReq.getRecallNum();
			recallNum = Math.min(recallNum, result.size());
			result = result.subList(0, recallNum);

			// 处理内容填充并过滤无效文档
			processContent(result, recallReq.getSceneId());

			log.info("RecallFacadeImpl recall result: {}", result);

			return result;
		} catch (Exception e) {
			log.error("Recall process failed for tenant: {}", recallReq.getTenantId(), e);
			return Collections.emptyList();
		} finally {
			TenantUtil.clear();
		}
	}

	private CompletableFuture<List<RecallResDTO>> processLibrariesAsync(String content, List<String> libraryIds, String tenantId) {
		if (CollectionUtil.isEmpty(libraryIds)) {
			return CompletableFuture.completedFuture(Collections.emptyList());
		}
		return CompletableFuture.supplyAsync(() -> {
			try {
				TenantUtil.setCurrentTenantId(tenantId);
				List<LibraryInfoDTO> libraries = libraryService.listLibraryByIds(libraryIds);
				List<String> enabledIds = libraries.stream()
						.filter(LibraryInfoDTO::getEnable)
						.map(LibraryInfoDTO::getId)
						.collect(Collectors.toList());
				return recallDataWithCache(content, enabledIds);
			} finally {
				TenantUtil.clear();
			}
		}, executor);
	}

	private List<RecallResDTO> recallDataWithCache(String query, List<String> libraryIds) {
		String cacheKey = "recall:cache:" + query.hashCode() + ":" + libraryIds.hashCode();
		String cached = redisUtil.get(cacheKey,String.class);
		if (cached != null) {
			return JSONUtil.toList(cached, RecallResDTO.class);
		}

		List<RecallResDTO> results = recallData(query, libraryIds);
		redisUtil.set(cacheKey, JSONUtil.toJsonStr(results), 60); // 缓存1分钟
		return results;
	}

	private List<RecallResDTO> recallData(String query, List<String> libraryIds) {
		List<SearchArticleDTO> articles = Optional.ofNullable(milvusClient.search(query, libraryIds)).orElseGet(Collections::emptyList);
		List<SearchDocDTO> docs = Optional.ofNullable(milvusClient.searchFile(query, libraryIds)).orElseGet(Collections::emptyList);

		return Stream.concat(
						articles.stream().map(ArticleCovert::covertToRecallRes),
						docs.stream().map(DocumentCovert::covertToRecallRes)
				).sorted(Comparator.comparingDouble(RecallResDTO::getConfidence).reversed())
				.collect(Collectors.toList());
	}

	private void processContent(List<RecallResDTO> results,String sceneId) {
		Iterator<RecallResDTO> iterator = results.iterator();
		while (iterator.hasNext()) {
			RecallResDTO dto = iterator.next();
			TenantUtil.setCurrentTenantId(dto.getTenantId());
			try {
				if (SourceEnum.ARTICLE.getCode().equals(dto.getSource())) {
					String cacheKey = "article:" + dto.getSourceId();
					String content = (String) redisUtil.get(cacheKey);
					if (content == null) {
						CsArticle article = articleService.getById(dto.getSourceId());

						if (article == null){
							iterator.remove();
							continue;
						}

						//更新热门问题缓存
						redisUtil.addToZset(sceneId,dto.getSourceId()+"##"+article.getTitle());

						// 更新缓存
						redisUtil.set(cacheKey, article.getContent(), CACHE_EXPIRE_TIME);
					}
					dto.setContent(content);
				} else if (SourceEnum.DOC.getCode().equals(dto.getSource())) {
					if (!processDocContent(dto)) {
						iterator.remove();
					}
				}
			} catch (Exception e) {
				log.error("Error processing content for {}", dto.getSourceId(), e);
				iterator.remove();
			}finally {
				TenantUtil.clear();
			}
		}
	}

	private boolean processDocContent(RecallResDTO dto) {

		String cacheKey = "doc:" + dto.getSourceId();
		CsDocuments byId = redisUtil.get(cacheKey, CsDocuments.class);
		if (byId == null) {
			byId = documentsService.getById(dto.getSourceId());
			redisUtil.set(cacheKey, byId, CACHE_EXPIRE_TIME);
		}

		if (byId == null) {
			// 移除
			return false;
		}
		dto.setContent(byId.getContent());

		String metadata = byId.getMetadata();

		DocumentDTO.MetaDataDTO bean = JSONUtil.toBean(metadata, DocumentDTO.MetaDataDTO.class);

		Map<String, String> metadataMap = new HashMap<>();
		metadataMap.put("title", bean.getTitle());
		metadataMap.put("pageSize", String.valueOf(bean.getPageSize()));
		metadataMap.put("index", String.valueOf(bean.getIndex()));
		metadataMap.put("content", dto.getContent());
		metadataMap.put("fileId", String.valueOf(byId.getFileId()));
		metadataMap.put("documentId", dto.getSourceId());

		dto.setMetadata(metadataMap);

		return true;
	}
}